package com.bigdata.kafka.utils;


import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.MissingArgumentException;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.Properties;

/**
 * Kafka 数据文件发送工具
 *
 * @author xin.ding
 * @date 2021/4/7 13:49
 */
public class SendFile {
    private static final Logger logger = LoggerFactory.getLogger(SendFile.class);
    private static String bootstrapServer = null;
    private static String topic = null;
    private static String file = null;
    private static int delayTime = 0;
    private static String key = null;
    private static boolean isPrint = false;
    /**
     * SASL 认证
     */
    private static boolean isSasl = false;
    private static String saslUsername = null;
    private static String saslPassword = null;
    /**
     * Kerberos 认证
     */
    private static boolean isKerberos = false;
    private static String krb5Conf = null;
    private static String jaasConf = null;
    private static final FastDateFormat DATE_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss.SSS");

    /**
     * 初始化配置
     */
    private static String[] buildRunArgs() {
        String f = "-";
        return new String[]{
                f + CliConstant.BOOTSTRAP_SERVERS, "10.243.141.133:9092",
                f + CliConstant.TOPIC, "test3",
                f + CliConstant.FILE, "D:\\code\\library\\util\\utils-all\\utils-bigdata\\src\\main\\resources\\config.properties",
                f + CliConstant.IS_PRINT,
                f + CliConstant.DELAY, "10000",
//                f + CliConstant.IS_DEBUG,
//                f + CliConstant.IS_HELP
        };

    }

    public static void main(String[] args) throws Exception {
        if (args.length == 0) {
            args = buildRunArgs();
        }
        parseArgs(args);

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class.getName());
        props.put(ProducerConfig.LINGER_MS_CONFIG, "5");
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        if (isKerberos) {
            KafkaAuth.buildKerberosEnv(props, krb5Conf, jaasConf);
        }
        if (isSasl) {
            KafkaAuth.buildKafkaSaslEnv(props, saslUsername, saslPassword);
        }

        try (Producer<String, String> producer = new KafkaProducer<>(props);
             BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                ProducerRecord<String, String> msg;
                if (null != key) {
                    msg = new ProducerRecord<>(topic, key, line);
                } else {
                    msg = new ProducerRecord<>(topic, line);
                }
                producer.send(msg);
                if (isPrint) {
                    System.out.println(String.format("[%s]: %s", DATE_FORMAT.format(System.currentTimeMillis()), line));
                }
                if (delayTime > 0) {
                    Thread.sleep(delayTime);
                }
            }
        } catch (IOException e) {
            logger.error("data send error", e);
        }
    }

    private static void parseArgs(String[] args) throws ParseException {
        CommandLineParser parser = new DefaultParser();
        Options options = new Options();

        options.addOption(CliConstant.BOOTSTRAP_SERVERS_SHORT, CliConstant.BOOTSTRAP_SERVERS, true, "bootstrap-server");
        options.addOption(CliConstant.TOPIC_SHORT, CliConstant.TOPIC, true, "topic");
        options.addOption(CliConstant.FILE_SHORT, CliConstant.FILE, true, "data file");
        options.addOption(CliConstant.DELAY_SHORT, CliConstant.DELAY, true, "delay time");
        options.addOption(CliConstant.IS_PRINT_SHORT, CliConstant.IS_PRINT, false, "print message");
        options.addOption(CliConstant.KEY_SHORT, CliConstant.KEY, true, "message key");
        options.addOption(CliConstant.SASL_USERNAME_SHORT, CliConstant.SASL_USERNAME, true, "sasl-username");
        options.addOption(CliConstant.SASL_PASSWORD_SHORT, CliConstant.SASL_PASSWORD, true, "sasl-password");
        options.addOption(CliConstant.KRB5_CONFIG_SHORT, CliConstant.KRB5_CONFIG, true, "krb5-conf");
        options.addOption(CliConstant.JAAS_CONFIG_SHORT, CliConstant.JAAS_CONFIG, true, "jaas-conf");
        options.addOption(CliConstant.IS_DEBUG_SHORT, CliConstant.IS_DEBUG, false, "debug");
        options.addOption(CliConstant.IS_HELP_SHORT, CliConstant.IS_HELP, false, "help info");

        CommandLine commandLine = parser.parse(options, args, true);

        if (commandLine.hasOption(CliConstant.IS_HELP)) {
            HelpFormatter formatter = new HelpFormatter();
            formatter.printHelp("use", options);
            System.exit(-1);
        }

        if (commandLine.hasOption(CliConstant.BOOTSTRAP_SERVERS)) {
            bootstrapServer = commandLine.getOptionValue(CliConstant.BOOTSTRAP_SERVERS);
        }
        if (commandLine.hasOption(CliConstant.TOPIC)) {
            topic = commandLine.getOptionValue(CliConstant.TOPIC);
        }
        if (commandLine.hasOption(CliConstant.FILE)) {
            file = commandLine.getOptionValue(CliConstant.FILE);
        }
        if (commandLine.hasOption(CliConstant.DELAY)) {
            delayTime = Integer.parseInt(commandLine.getOptionValue(CliConstant.DELAY).trim());
        }
        if (commandLine.hasOption(CliConstant.IS_PRINT)) {
            isPrint = true;
        }
        if (commandLine.hasOption(CliConstant.KEY)) {
            key = commandLine.getOptionValue(CliConstant.KEY);
        }
        if (commandLine.hasOption(CliConstant.SASL_USERNAME) && commandLine.hasOption(CliConstant.SASL_PASSWORD)) {
            isSasl = true;
            saslUsername = commandLine.getOptionValue(CliConstant.SASL_USERNAME);
            saslPassword = commandLine.getOptionValue(CliConstant.SASL_PASSWORD);
        }
        if (commandLine.hasOption(CliConstant.KRB5_CONFIG) && commandLine.hasOption(CliConstant.JAAS_CONFIG)) {
            isKerberos = true;
            krb5Conf = commandLine.getOptionValue(CliConstant.KRB5_CONFIG);
            jaasConf = commandLine.getOptionValue(CliConstant.JAAS_CONFIG);
        }
        if (commandLine.hasOption(CliConstant.IS_DEBUG)) {
            delayTime = 1000;
            isPrint = true;
        }
        if (commandLine.hasOption(CliConstant.IS_DEBUG)) {
            delayTime = 1000;
            isPrint = true;
        }

        if (StringUtils.isAnyEmpty(bootstrapServer, topic, file)) {
            System.err.println("bootstrap-server or topic or file can not be null");
            throw new MissingArgumentException("miss argument");
        }
    }
}
